package com.pie4cloud.pie.bus.api.consumer;

import com.alibaba.fastjson.JSON;
import com.pie4cloud.pie.bus.api.dto.ConsumerDto;
import com.pie4cloud.pie.bus.api.dto.MessageDto;
import com.pie4cloud.pie.bus.api.consumer.factory.IKfkFactory;
import com.pie4cloud.pie.common.core.util.R;
import com.pie4cloud.pie.common.security.annotation.Inner;
import io.swagger.annotations.Api;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@RestController()
@RequestMapping("/kfk")
@Api(value = "kfk", tags = "kafka消息管理")
public class ConsumerController {
	private String fileDirector="";
	@Value("${spring.application.name}")
	private String applicationName;
	@Autowired
	private SnowFlake snowFlake;

	@Autowired
	private MetricCalcService metricCalcService;
	@Autowired
	private IKfkFactory iKfkFactory;

	private ThreadPoolExecutor pool = new ThreadPoolExecutor(50, 1000, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(1), Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());

	/*private ThreadPoolExecutor pool1 = new ThreadPoolExecutor(200, // corSize
			1000, // maxPoolSize
			60, // keepAliveTime
			TimeUnit.SECONDS, // unit,
			new ArrayBlockingQueue(1),// workQueue
			new ThreadPoolExecutor.AbortPolicy()

	);
*/

	@PostConstruct
	public void init() {
		String property = System.getProperty("user.dir");
		fileDirector=property+"/cache/"+applicationName;
		File file = new File(fileDirector);
		if (!file.exists() || !file.isDirectory()) {
			FileUtils.deleteQuietly(file);
			boolean bool1 = (new File(this.fileDirector)).mkdirs();
		}
		File[] fileList = file.listFiles();
		if(null!=fileList&& fileList.length>0){
			for(File fileE:fileList){
				pool.execute(()->{
					String content=this.readFileContent(fileE);
					ConsumerDto consumerDto= JSON.parseObject(content,ConsumerDto.class);
					IKfkaConsumer iKfkaConsumer=iKfkFactory.get(consumerDto.getHandler());
					try {
						iKfkaConsumer.consume(consumerDto.getContent(),consumerDto.getTopic(),consumerDto.getKafGroup());
					} catch (Exception e) {
						List<MessageDto> messageDtos=new ArrayList<>();
						MessageDto messageDto=new MessageDto();
						messageDto.setContent(consumerDto.getContent());
						messageDto.setTopic(consumerDto.getTopic());
						messageDtos.add(messageDto);
						metricCalcService.submit(messageDtos);
					}finally {
						FileUtils.deleteQuietly(fileE);
					}
				});

			}

		}

	}
	public  String readFileContent(File file) {
		BufferedReader reader = null;
		StringBuffer sbf = new StringBuffer();
		try {
			reader = new BufferedReader(new FileReader(file));
			String tempStr;
			while ((tempStr = reader.readLine()) != null) {
				sbf.append(tempStr);
			}
			reader.close();
			return sbf.toString();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (reader != null) {
				try {
					reader.close();
				} catch (IOException e1) {
					e1.printStackTrace();
				}
			}
		}
		return sbf.toString();
	}
	public void writerFile(String content, String fileName) {
		FileChannel channel = null;
		FileLock lock = null;
		RandomAccessFile raf = null;
		try {
			raf = new RandomAccessFile(fileName, "rw");
			raf.seek(raf.length());
			channel = raf.getChannel();
			while (true) {
				lock = channel.tryLock();
				if (lock != null)
					break;
				Thread.sleep(500L);
			}
			ByteBuffer sendBuffer = ByteBuffer.wrap(content.getBytes("UTF-8"));
			channel.write(sendBuffer);
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			if (lock != null)
				try {
					lock.release();
				} catch (IOException e) {
					e.printStackTrace();
				}
			if (channel != null)
				try {
					channel.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			if (raf != null)
				try {
					raf.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
		}
	}

/*	@Inner
	@PostMapping("/consume")
	public R consume(@RequestBody ConsumerDto consumerDto) {
		try {
			String fileName=fileDirector+"/"+snowFlake.nextId()+".data";
			this.writerFile(JSON.toJSONString(consumerDto),fileName);
			pool.execute(()->{
				IKfkaConsumer iKfkaConsumer=iKfkFactory.get(consumerDto.getHandler());
				try {
					iKfkaConsumer.consume(consumerDto.getContent(),consumerDto.getTopic(),consumerDto.getKafGroup());
				} catch (Exception e) {
					List<MessageDto> messageDtos=new ArrayList<>();
					MessageDto messageDto=new MessageDto();
					messageDto.setContent(consumerDto.getContent());
					messageDto.setTopic(consumerDto.getTopic());
					messageDtos.add(messageDto);
					metricCalcService.submit(messageDtos);
				}finally {
					FileUtils.deleteQuietly(new File(fileName));
				}
			});
		} catch (Exception e) {
			return R.failed();
		}

		return R.ok();
	}*/

	@Inner
	@PostMapping("/consume")
	public R consume(@RequestBody ConsumerDto consumerDto) {
		R r =new R();
		try {
			String fileName = fileDirector + "/" + snowFlake.nextId() + ".data";
			this.writerFile(JSON.toJSONString(consumerDto), fileName);
			IKfkaConsumer iKfkaConsumer = iKfkFactory.get(consumerDto.getHandler());
			Callable<R> call = new Callable() {
				R result=new R();
				@Override
				public R call() throws Exception {
					for( int i=0;i<3;i++){
						try {
							result=iKfkaConsumer.consume(consumerDto.getContent(), consumerDto.getTopic(), consumerDto.getKafGroup());
							if(0==result.getCode()){
								break;
							}
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
					return result;
				}
			};
			FutureTask<R> futureTask=new FutureTask<>(call);
			pool.submit(futureTask);
			try {
				futureTask.get(3,TimeUnit.MINUTES);
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			} catch (TimeoutException e) {
				e.printStackTrace();
			}
			FileUtils.deleteQuietly(new File(fileName));
		} catch (Exception e) {
			return r.failed();
		}

		return r;
	}
}
